package dse;

import java.util.*;

import dse.messages.*;
import java.io.*;

public class Node implements Comparable<Node>, Serializable 
{
	private static final long serialVersionUID = 1L;
	public static final double diskAccessTime = 0.00004;
	public static final int InvalidLocalId = -1;
	public static int queuedRequestSize = 0;
	
	/** Unique global id of the node. */
	private int id;
	/** Local id of the node unique to its own cluster. */ 
	private int localId;
	
	/** Id of the cluster the node is in. */
	private int cluster;
	private boolean isSupernode;
	private NodeState state;
	
	/** This is the max bytes that can be uploaded or downloaded in a second. */
	private int bandwidthLimit;
	
	private Hashtable<Integer, DataShare> hostedDataShares;
	//add for svr
	private LinkedList<DataObject> svrinit;
	//end add for svr
	
	public static Integer currentSequence = 0;
	//public Hashtable<Version, Integer> recentUpdates; 
	public Hashtable<DataObject, TreeMap<Version, Integer>> dataVersions; 
	
	private double localTime;
	/**
	 * This is a collection of recent requests the node has received a local
	 * get data share message for.
	 */
	//private Map<DataObject,Integer> recentRequests;
	
	/**Add***/
	//private LinkedList<Pair<DataObject, Integer>> unactivatedUpdatedDataObjectShares;
	private int sequence;
	private Hashtable<Integer, TreeSet<DataRequest>> requestList;
	private Hashtable<Integer, TreeSet<DataRequest>> grantedList;
	private LinkedList<DataRequest> clientRequests;
	private LinkedList<DataRequest> leaderDecisions;
	
	
	public Node (int id) {
		this.id = id;
		this.localId = Node.InvalidLocalId;
		this.cluster = Cluster.InvalidId;
		this.isSupernode = false;
		this.state = NodeState.Invalid;
		
		this.bandwidthLimit = 0;
		//this.unactivatedUpdatedDataObjectShares = null;
		this.hostedDataShares = null;
		this.sequence = 0;
		this.localTime = 0;
		
		this.requestList = null;
		this.grantedList = null;
		this.clientRequests = null;
		this.leaderDecisions = null;
		
		this.currentLock = null;
		this.versionFile = null;		
		this.updateList = null;
	}
	
	public int id () { return id; } 
	public int localId () { return localId; }
	public void setLocalId (int newLocalId) { localId = newLocalId; }
	
	public int clusterId () { return cluster; }
	public void setClusterId (int newClusterId) { cluster = newClusterId; }
	
	public boolean isSupernode () { return isSupernode; }
	public void setIsSupernode (boolean status) { isSupernode = status; }
	
	public NodeState state () { return state; }
	public void setState (NodeState newState) { state = newState; }
	
	public boolean isUp () { return state == NodeState.Up; }
	public boolean isDown () { return state == NodeState.Down; }
	
	public int bandwidthLimit () { return bandwidthLimit; }
	public void setBandwidthLimit (int value) { bandwidthLimit = value; }
	public int sequence () {return ++sequence;}
	public int getSequence () {return sequence;}
	public void setSequence (int s) {sequence = s;}
	
	
	/**
	 * This method should be overriden if possible to provide a better distance
	 * calculation between nodes.
	 * 
	 * @param other
	 * @return
	 */
	public double distanceTo (Node other)
	{
		return 0;
	}
	
	public void host (DataObject dataObject)
	{
		/*if (this.hostedDataObjects == null)
			this.hostedDataObjects = new LinkedList<DataObject>();
		this.hostedDataObjects.add(dataObject);*/
	}
	
	public boolean isHosting (DataObject dataObject) {
		/*if (this.hostedDataObjects == null)
			return false;
		return this.hostedDataObjects.contains(dataObject);*/
		return false;
	}
	
	public void host (DataObject dataObject, int shareIndex) {
		DataShare ds = new DataShare(dataObject, shareIndex);
		if (this.hostedDataShares == null)
			this.hostedDataShares = new Hashtable<Integer, DataShare>();
		this.hostedDataShares.put(dataObject.id(), ds);
		if(shareIndex == 1){
			if(this.svrinit == null)
				this.svrinit = new LinkedList<DataObject>();
			this.svrinit.add(dataObject);
		}
	}
	
	public boolean isHosting (DataObject dataObject, int shareIndex) {
		/*if (this.hostedDataObjectShares == null)
			return false;
		return this.hostedDataObjectShares.contains(
				new Pair<DataObject, Integer>(dataObject, shareIndex)
				);*/
		if(this.hostedDataShares == null)
			return false;
		else if(this.hostedDataShares.keySet().contains(dataObject.id())
				&& this.hostedDataShares.get(dataObject.id()).getIndex()==shareIndex)
			return true;
		else
			return false;
	}
	
	public LinkedList<DataObject> getSVRInitiator(){
		return this.svrinit;
	}
	
	public DataShare getDataShare(DataObject dobj)
	{
		if (this.hostedDataShares == null)
			return null;
		return this.hostedDataShares.get(dobj.id());
	}
	
	public Collection<DataShare> getAllShares()
	{
		if (hostedDataShares == null)
			return null;
		return hostedDataShares.values();
	}
	
	public Hashtable<Integer, DataShare> getHostShares(){
		return hostedDataShares;
	}

	public int compareTo (Node other) {
		return this.id - other.id;
	}
	public int compareTo (int id) {
		return this.id - id;
	}
		
	
	/**
	 * Useful for debugging.
	 */
	@Override
	public String toString () {
		String res = String.format("node#%d/%d", new Object [] {id,cluster});
		if (isSupernode)
			return res + "($)";
		return res;
	}
	
	public void update (DataObject data, int shareIndex, boolean activate) {
		/*if (activate) 
		{
			if(!isHosting(data, shareIndex))
				host(data, shareIndex);
		} 
		else
		{
			if (this.unactivatedUpdatedDataObjectShares == null)
				this.unactivatedUpdatedDataObjectShares = 
					new LinkedList<Pair<DataObject,Integer>>();
			this.unactivatedUpdatedDataObjectShares.add(
					new Pair<DataObject,Integer>(data,shareIndex)
					);
		}*/
			
	}
	
	public boolean isHostingUnactive (DataObject data, int share) {
		/*if(this.unactivatedUpdatedDataObjectShares == null)
			return false;
		else
			return this.unactivatedUpdatedDataObjectShares.contains(new Pair<DataObject, Integer>(data, share));*/
		return false;
	}
	
	public void activateUpdatedDataObjectShare (DataObject data, int shareIndex) 
	{
		/*Pair<DataObject, Integer> share = new Pair<DataObject, Integer>(data, shareIndex);
		if (unactivatedUpdatedDataObjectShares.contains(share) == false)
			return;
		unactivatedUpdatedDataObjectShares.remove(share);
		if(!isHosting(data, shareIndex))
			host(data, shareIndex);*/
	}
	public void removeUnactivatedUpdatedDataObjectShare (DataObject data, int shareIndex) {
		//unactivatedUpdatedDataObjectShares.remove( new Pair<DataObject, Integer>(data, shareIndex));
	}
	public void removeHostedDataObjectsAndShares()
	{
		/*if(this.hostedDataObjects != null)
			this.hostedDataObjects.clear();
		if(this.hostedDataObjectShares != null)
			this.hostedDataObjectShares.clear();*/
	}
	public void removeCachedDataObjectsAndShares()
	{
		/*if(this.cachedDataObjects != null)
			this.cachedDataObjects.clear();
		if(this.cachedDataObjectShares != null)
			this.cachedDataObjectShares.clear();*/
	}
	
	public static int newSequence()
	{
		synchronized (currentSequence){
			currentSequence++;
			return currentSequence.intValue();
		}
	}
	//modified for two-level
	public void updateSuccess(Version version, DataObject dobj)
	{
		if(dataVersions == null)
			dataVersions = new Hashtable<DataObject, TreeMap<Version, Integer>>();
		if(dataVersions.get(dobj) == null)
			dataVersions.put(dobj, new TreeMap<Version, Integer>());
		if(dataVersions.get(dobj).get(version) == null)
			dataVersions.get(dobj).put(version, 0);
		dataVersions.get(dobj).put(version, dataVersions.get(dobj).get(version)+1);
	}
	public Version getMaxVersion(DataObject dobj, boolean ru, int threshold){
		if(dataVersions == null)
			return DataShare.initVersion;
		else if(dataVersions.get(dobj) == null)
			return DataShare.initVersion;
		else if(dataVersions.get(dobj).size() == 0)
			return DataShare.initVersion;
		else{
			if(ru)
				return dataVersions.get(dobj).lastKey();
			else{
				Version max = this.dataVersions.get(dobj).lastKey();
				while(max!=null && this.dataVersions.get(dobj).get(max).intValue() < threshold){
					max = this.dataVersions.get(dobj).lowerKey(max);
				}
				if(max != null)
					return max;
				else
					return DataShare.initVersion;
			}
		}
	}
	private Hashtable<DataObject, TreeMap<Integer, Version>> curReads;
	public void addCurReads(DataObject dobj, int i, Version v){
		if(this.curReads == null)
			this.curReads = new Hashtable<DataObject, TreeMap<Integer, Version>>();
		if(this.curReads.get(dobj) == null)
			this.curReads.put(dobj, new TreeMap<Integer, Version>());
		this.curReads.get(dobj).put(i, v);
	}
	public Version removeCurReads(DataObject dobj, int i){
		//if(this.curReads == null)
		//	this.curReads = new Hashtable<DataObject, TreeMap<Integer, Version>>();
		//if(this.curReads.get(dobj) == null)
		//	this.curReads.put(dobj, new TreeMap<Integer, Version>());
		
		this.curReads.get(dobj).remove(i);
		Version v = null;
		if(this.curReads.get(dobj).size() > 0)
			v = this.curReads.get(dobj).firstEntry().getValue();
		Version bv = null;
		if(this.dataVersions != null){
			TreeMap<Version, Integer> vi = this.dataVersions.get(dobj);
			if(vi != null){
				if(v == null){
					bv = vi.lastKey();
					while((bv!=null) && (vi.get(bv)<dobj.shareCount())){
						bv = vi.lowerKey(bv);
					}
				}
				else{
					Version v2 = vi.firstKey();
					while((v2!=null) && (v2.compareTo(v)<0)){
						if(vi.get(v2) >= dobj.shareCount()){
							bv = v2;
						}
						v2 = vi.higherKey(v2);
					}
				}
				if(bv != null){
					while(vi.size()>1 && vi.firstKey().compareTo(bv)<0)
						vi.pollFirstEntry();
				}
			}
		}
		
		return bv;
	}
	
	///End two level order
	
	
	//process message
	public void enqueueMessage(Message m, ExperimentRecorder recorder)
	{
		if(m.endTime() < this.localTime){
			this.localTime += Node.diskAccessTime;
			//System.out.println("need more time");
		}
		else
			this.localTime = m.endTime()+Node.diskAccessTime;
		
		m.deliverTime(this.localTime);
		//m.deliverTime(m.endTime());
		m.isFinished(true);
	}
	
	//Lock related
	public TreeSet<DataRequest> getQueuedRequestList(int did){
		if(this.requestList==null)
			return null;
		return this.requestList.get(did);
	}
	
	public TreeSet<DataRequest> getGrantedRequestList(int did){
		if(this.grantedList == null)
			return null;
		return this.grantedList.get(did);
	}
	
	public boolean receiveClientRequest(DataRequest request){
		if((this.leaderDecisions!=null) && (this.leaderDecisions.contains(request))){
			this.leaderDecisions.remove(request);
			request.setFollowerReady(true);
			return true;
		}
		else{
			this.addClientRequest(request);
			return false;
		}
	}
	
	public boolean receiveLeaderDecision(DataRequest request){
		assert request.getLeader().id() != this.id();
		if((this.clientRequests!=null) && (this.clientRequests.contains(request))){
			this.clientRequests.remove(request);
			request.setFollowerReady(true);
			return true;
		}
		else{
			addLeaderDecision(request);
			return false;
		}
	}
	
	public boolean canLock(int did, DataRequest req){	
		if((this.requestList!=null) && (this.requestList.get(did)!=null) && (this.requestList.get(did).size() > 0) 
				&& (this.requestList.get(did).first().getTranxID() != req.getTranxID()))
        	return false;
       	
		if((this.grantedList==null) || (this.grantedList.get(did)==null) || (this.grantedList.get(did).size()==0))
			return true;
		
		if((req.getType() == DataRequest.READ) && (this.grantedList.get(did).first().getType()==DataRequest.READ)){
			return true;
		}
		if((req.getType() == DataRequest.WRITE) && (this.grantedList.get(did).first().getType()==DataRequest.WRITE)){
			return true;
		}
		
		return false;
	}
	
	public void addGrantedRequest(int did, DataRequest request){
		if(this.grantedList == null)
			this.grantedList = new Hashtable<Integer, TreeSet<DataRequest>>();
		if(this.grantedList.get(did) == null)
			this.grantedList.put(did, new TreeSet<DataRequest>());
		
		this.grantedList.get(did).add(request);
	}
	
	public void addQueuedRequest(int did, DataRequest request){
		if(this.requestList == null)
			this.requestList = new Hashtable<Integer, TreeSet<DataRequest>>();
		if(this.requestList.get(did) == null)
			this.requestList.put(did, new TreeSet<DataRequest>());
		
		this.requestList.get(did).add(request);
		if(this.requestList.get(did).size() > Node.queuedRequestSize)
			Node.queuedRequestSize = this.requestList.get(did).size();
	}
	
	private void addClientRequest(DataRequest request){
		if(this.clientRequests == null)
			this.clientRequests = new LinkedList<DataRequest>();
		
		this.clientRequests.add(request);
	}
	
	private void addLeaderDecision(DataRequest request){
		if(this.leaderDecisions == null)
			this.leaderDecisions = new LinkedList<DataRequest>();
		
		this.leaderDecisions.add(request);
	}
	
	public void releaseLock(int did, DataRequest request){
		if((this.grantedList!=null) && (this.grantedList.get(did)!=null))
			this.grantedList.get(did).remove(request);
	}
		
	public void removeOldShare(){
		for(DataShare ds : this.hostedDataShares.values()){
			ds.removeOldVersion();
		}
	}
	///////////
	//ADD NEW FOR REPERASURE//
	///////////
	private Hashtable<Integer, Boolean> currentLock;
	private Hashtable<Integer, TreeMap<Integer, Boolean>> versionFile;
	private Hashtable<Integer, LinkedList<DataRequest>> updateList;
	
	public boolean canIssueUpdate(int did){
		if(this.currentLock != null){
			Boolean lock = this.currentLock.get(did);
			if(lock == null){
				return true;
			}
			else{
				return !lock.booleanValue();
			}
		}
		else{
			this.currentLock = new Hashtable<Integer, Boolean>();
			return true;
		}
	}
	public boolean updateRequest(int did, DataRequest req){
		if(canIssueUpdate(did)){
			this.currentLock.put(did, true);
			return true;
		}
		else{
			if(this.updateList == null){
				this.updateList = new Hashtable<Integer, LinkedList<DataRequest>>();
				this.updateList.put(did, new LinkedList<DataRequest>());
			}
			else if(this.updateList.get(did) == null){
				this.updateList.put(did, new LinkedList<DataRequest>());
			}
			this.updateList.get(did).add(req);
			return false;
		}
	}
	public DataRequest getNextUpdate(int did){
		if((this.updateList != null) && (this.updateList.get(did) != null)
				&& this.updateList.get(did).size() > 0){
			return this.updateList.get(did).getFirst();
		}
		return null;
	}
	public void ReleaseUpdate(int did){
		if(this.currentLock != null)
			this.currentLock.remove(did);
	}
	public void LockUpdate(int did){
		if(this.currentLock == null)
			this.currentLock = new Hashtable<Integer, Boolean>();
		this.currentLock.put(did, true);
	}
	public int getMaxVersion(int did, boolean ru){
		if(this.versionFile == null){
			this.versionFile = new Hashtable<Integer, TreeMap<Integer, Boolean>>();
			this.versionFile.put(did, new TreeMap<Integer, Boolean>());
			this.versionFile.get(did).put(0, false);
		}
		else if(this.versionFile.get(did) == null){
			this.versionFile.put(did, new TreeMap<Integer, Boolean>());
			this.versionFile.get(did).put(0, false);
		}
		
		if(ru){
			return this.versionFile.get(did).lastKey().intValue();
		}
		else{
			Integer max = this.versionFile.get(did).lastKey();
			while(max!=null && this.versionFile.get(did).get(max).booleanValue()){
				max = this.versionFile.get(did).lowerKey(max);
			}
			if(max != null)
				return max.intValue();
			else
				return 0;
		}
	}
	public void writeVersion(int did, int version, boolean on){
		if(this.versionFile == null){
			this.versionFile = new Hashtable<Integer, TreeMap<Integer, Boolean>>();
			this.versionFile.put(did, new TreeMap<Integer, Boolean>());
			this.versionFile.get(did).put(0, false);
		}
		else if(this.versionFile.get(did) == null){
			this.versionFile.put(did, new TreeMap<Integer, Boolean>());
			this.versionFile.get(did).put(0, false);
		}
		
		this.versionFile.get(did).put(version, on);
		if(!on){
			Integer max = this.versionFile.get(did).lowerKey(version);
			while(max != null){
				this.versionFile.get(did).remove(max);
				max = this.versionFile.get(did).lowerKey(version);
			}
		}
	}
}
